package david.java.flink_sql.streamApiIntegration;


import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Description:
 * @See https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/data_stream_api/#examples-for-createtemporaryview
 * @Author: ZhaoDawei
 * @Date: Create in 10:14 上午 2022/3/29
 */
public class S3_2_CreateTemporaryView {


    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStreamSource<Tuple2<Long, String>> dataStream = env.fromElements(
                Tuple2.of(12L, "Alica"),
                Tuple2.of(0L, "Bob"));


        // === EXAMPLE 1 ===

        // register the DataStream as view "MyView" in the current session
        // all columns are derived automatically
        tableEnv.createTemporaryView("MyView", dataStream);
        tableEnv.from("MyView").printSchema();
        // (
        //   `f0` BIGINT NOT NULL,
        //   `f1` STRING
        // )


        // === EXAMPLE 2 ===

        // register the DataStream as view "MyView" in the current session,
        // provide a schema to adjust the columns similar to `fromDataStream`

        // in this example, the derived NOT NULL information has been removed
        tableEnv.createTemporaryView(
                "MyView2",
                dataStream,
                Schema.newBuilder()
                .column("f0", "BIGINT")
                .column("f1", "String")
                .build());
        tableEnv.from("MyView2").printSchema();
        // (
        //   `f0` BIGINT,
        //   `f1` STRING
        // )


        // === EXAMPLE 3 ===

        // use the Table API before creating the view if it is only about renaming columns
        tableEnv.createTemporaryView(
                "MyView3",
                tableEnv.fromDataStream(dataStream).as("id", "name"));
        tableEnv.from("MyView3").printSchema();
        // (
        //   `id` BIGINT NOT NULL,
        //   `name` STRING
        // )

    }
}
